/*
 * TCSS 558
 * Distributed HashMap
 * Fall 2013
 * James Marquardt
 * Matt Adams
 * Justin Wong
 */
package node;

import static com.google.common.base.Preconditions.*;

import java.io.FileNotFoundException;
import java.rmi.AccessException;
import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import node.tasks.FixFingerTable;
import node.tasks.Stabilize;

import model.Entries;
import model.Entry;
import model.HostPort;
import model.NodeReferences;
import model.NodesAndEntries;

import registry.NodeLocationException;
import registry.RegistrationPacket;
import registry.RemoteRegistry;
import tools.CheckArgument;
import tools.Log;

/**
 * Class represents actual nodes in a chord peer-to-peer network.
 * @author Matt Adams
 */
public class NodeActual extends Node {
    
    /**
     * Start delay before stabilize begins.
     */
    private static int STABILIZE_START_DELAY = 15;
    
    /**
     * Time between stabilize task execution.
     */
    private static int STABILIZE_INTERVAL = 15;
    
    /**
     * Start delay before fix finger table begins.
     */
    private static int FIX_FINGER_START_DELAY = 2;
    
    /**
     * Time between fix finger table task execution.
     */
    private static int FIX_FINGER_INTERVAL = 12;
    
    /**
     * The RemoteNodeRMIEndpoint attached to this NodeActual.
     */
    private RemoteNodeRMIEndpoint myEndpoint = null;
    
    /**
     * Host and port of an arbitrary node registered in the chord.
     */
    private HostPort initialHostPortReference;
    
    /**
     * NodeRMIProxy object created from the initialHostPortReference.
     */
    private NodeRMIProxy initialNodeReference = null;
    
    /**
     * m value determines max number of nodes as being 2^m.
     */
    protected int mValueOfChord;
    
    /**
     * Routing table (including finger table, and successor/predecessor
     * references)
     */
    protected NodeReferences references;

    /**
     * Repository for locally stored entries (Like virtual nodes).
     */
    private Entries entries;
    
    /**
     * NodeRMIProxy object representing this NodeActual.
     */
    private Node myProxy;
    
    /**
     * Executor service for local node tasks.
     */
    private ScheduledExecutorService nodeTasks;
    
    /**
     * The remote node registry.
     */
    private RemoteRegistry remoteService;
    
    /**
     * Constructs a NodeActual object.
     * 
     * @param remoteService Remote service registry.
     * @param hostPort Host and port.
     */
    public NodeActual(final RemoteRegistry remoteService, final HostPort hostPort) {
        this(remoteService, hostPort, null);
    }
    
    /**
     * Constructs a NodeActual object.
     * 
     * @param remoteService Remote service registry.
     * @param hostPort Host and port.
     * @param entries Initial entries.
     */
    public NodeActual(final RemoteRegistry remoteService,
                      final HostPort hostPort,
                      final Entries entries) {
        checkNotNull(remoteService, "The reference to the remote node registry cannot be null.");
        this.nodeHostPort = checkNotNull(hostPort, "HostPort of this Node cannot be null");
        this.remoteService = remoteService;
        try {
            RegistrationPacket packet = remoteService.registerNode(this.nodeHostPort);  
            this.initialHostPortReference = packet.initialNodeReference;
            this.mValueOfChord = packet.m_value;
            this.nodeID = packet.node_id;
            this.messageToFile("Node " + this.nodeID + " created");
            
            if (entries == null) {
                this.entries = new Entries(this.mValueOfChord);
            } else {
                this.entries = entries;
            }
            this.references = new NodeReferences(nodeID, hostPort, entries, mValueOfChord);
            this.nodeTasks = new ScheduledThreadPoolExecutor(2, new NodeThreadFactory("NodeTaskExecution:" + this.nodeHostPort.toString()));
            
            this.myEndpoint = RemoteNodeRMIEndpoint.createRMIEndpoint(this, hostPort);
            this.myEndpoint.openConnections(); 
            this.myProxy = new NodeRMIProxy(this.nodeHostPort, this.nodeHostPort);
            setInitialNodeReferences();
            startTasks();
            
        } catch (RemoteException e) {
            throw new RuntimeException(e);
        } catch (NodeLocationException e) {
            e.printStackTrace();
        }  catch (NotBoundException e) {
            e.printStackTrace();
        }
    }
    
    /**
     * Helper to initialize this NodeActual into a Chord via an arbitrary registered node reference.
     */
    private void setInitialNodeReferences() {
        try {
            if (this.initialHostPortReference == null) {
                this.references.setSuccessor(this.myProxy);               
            } else {
                this.initialNodeReference = new NodeRMIProxy(this.nodeHostPort, this.initialHostPortReference);
                NodeRMIProxy mySuccessor = (NodeRMIProxy) this.initialNodeReference.findSuccessor(this.nodeID);
                if (mySuccessor.nodeID == this.nodeID) {
                    this.references.setSuccessor(this.initialNodeReference);
                } else {
                    this.references.setSuccessor(mySuccessor);
                }   
            }
            messageToFile("Initial references set:");
            for(Node n : this.references.getFingerTable().getFullFingerTable()) {
            	if(n != null) {
            		messageToFile(Integer.toString(n.getNodeID()) + " " + n.getNodeHostPort());
            	}
            }
            messageToFile("Successor: " + this.references.getSuccessor().getNodeID() +" " + this.references.getSuccessor().getNodeHostPort());
        } catch (AccessException e) {
            e.printStackTrace();
        } catch (RemoteException e) {
            e.printStackTrace();
        } catch (NotBoundException e) {
            e.printStackTrace();
        }
    }
    
    @Override
    public Node findSuccessor(int key) {
        if (key < 0 && key < (int) Math.pow(2, this.mValueOfChord)) {
            throw new IllegalArgumentException("Node ID should be greater than 0 and less than " + 
                                                (int) Math.pow(2, this.mValueOfChord) + " (2^m). Received ID: " + key);
        }

        Log.out("Finding successor for key ID: " + key + ". Local node ID: " + this.nodeID);
        Node successor = this.references.getSuccessor();
        Node result = null;
        //Check for null successor, should only occur when node is only one in the network.
        if (successor == null ) {
            Log.out("NodeID: " + this.nodeID + " has no successor, returning itself as its own successor");
            return this.myProxy;
        }
        //Check if the key is in the interval of this node ID and the successor's node ID.
        if (CheckArgument.isInInterval(key, this.nodeID, successor.getNodeID())
                || key == successor.getNodeID()) {
            result = successor;
            Log.out("Key ID: " + key + " is in interval(" + this.nodeID + ", " + successor.getNodeID() + ")");
        } else {  
            Log.out("Key ID: " + key + " is NOT in interval(" + this.nodeID + ", " + successor.getNodeID() + ")");
            //Request closest preceding node from references and call findSuccessor with the provided key
            Node closestPrecedingNode = this.references.getClosestPrecedingNode(key);
            result = closestPrecedingNode.findSuccessor(key);
        }

        Log.out("Successor Found. Successor ID: " + result.getNodeID());
        return result;
    }

    @Override
    public void insertEntry(Entry entry) {
        final int key= getNodeIDFromKey(entry.getID(), this.mValueOfChord);
        final Set<Entry> entrieCopies = new HashSet<Entry>();
    	try {
    		int start = this.getNodeID();
    		int end = this.getNodeReferences().getPredecessor().getNodeID();

    		if(CheckArgument.isInInterval(key, end, start)) {
    			System.out.println("in interval of " + start);
    			messageToFile("Inserting Entry <"+entry.getID() +", " +entry.getValue()+"> to virtual " + key);
    			this.entries.add(entry, key);
    			entrieCopies.add(entry);
    			this.references.getSuccessor().insertCopies(entrieCopies);
    		} else {
    		    Node foundNode = this.findSuccessor(key);
    		    foundNode.insertEntry(entry);
    		    entrieCopies.add(entry);
    		    foundNode.insertCopies(entrieCopies);
    		}
    	} catch (NullPointerException e) {
    		//Log.err("Cannot guarantee insert on non stable system. Wait for system to stabilize.");
    		messageToFile("Entry inserted on non stable system. System must be stable for entry retrieval to be guaranteed.");
    		
    		this.entries.add(entry, key);
    	}
    }

    @Override
    public void removeEntry(Entry entry) {
        this.entries.remove(entry);

        final Set<Entry> entriesToRemove = new HashSet<Entry>();
        entriesToRemove.add(entry);
        this.references.getSuccessor().removeCopies(entriesToRemove);
        
    }

    @Override
    public boolean containsEntry(Entry entry) {
        // TODO Auto-generated method stub
        return false;
    }
    
    @Override
    public synchronized Entry getEntry(int the_key) {
    	try {
    		int key = getNodeIDFromKey(the_key, this.mValueOfChord);
    		int start = this.getNodeID();
    		int end = this.getNodeReferences().getPredecessor().getNodeID();

    		if(CheckArgument.isInInterval(key, end, start)) {

    			for(Entry e : entries.getEntries(key)) {
    				//System.out.println("Checking entries in node " + this.getNodeID());
    				//System.out.println(e.getValue());
    				if(e.getID() == the_key) {
    					messageToFile("Get request for key " + the_key + " found record <"+e.getID()+", "+e.getValue()+">");
    					return e;
    				}
    			}
    			messageToFile("Get request for key " + the_key + " found no record.");
    			return null;
    			//return new Entry(1, "default");
    		} else {
    			messageToFile("Forwarding get of key hashed " + key + " to node " + this.getNodeReferences().getClosestPrecedingNode(key).getNodeID());
    			return this.getNodeReferences().getClosestPrecedingNode(key).getEntry(the_key);
    		}
    	} catch (NullPointerException e) {
    		Log.err("Cannot gurantee get on unstable system. Wait for system to stabilize");
    		messageToFile("Cannot gurantee get on unstable system.");
    		return null;
    	}

    }

    @Override
    public void leaveNetworkNotify(Node predecessor) {
    	Log.out("Node LEAVING network, nodeID: " + predecessor.nodeID);
        Log.out("Removing reference with nodeID: " + predecessor.nodeID + " from nodeID: " + this.nodeID);
    	this.references.removeReference(predecessor);
    	predecessor.disconnect();
    }
    
    @Override
    public void addAllEntries(Set<Entry> allEntries) {
    	this.entries.addAll(allEntries);
    }
    
    @Override
    public void disconnect() {
    	messageToFile("Node " + this.getNodeID() + " leaving network");
    	messageToFile("Shutting down tasks");
    	try {
			this.remoteService.removeNode(this.getNodeID());
		} catch (RemoteException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
        List<Runnable> tasks = this.nodeTasks.shutdownNow();
        
        Log.out("Shutdown tasks for node: " + this.nodeID + " Number of tasks shutdown: " + tasks.size());
        messageToFile("Node successfully left network");
    }
    
    @Override
    public NodesAndEntries notifyWithPotentialPredecessor(Node potentialPredecessor) {
        // the result will contain this node's predecessor and successor.
        final List<Node> nodes = new LinkedList<Node>();

        //Add references to predecessor and successor
        if (this.references.getPredecessor() != null) {
            //Check if predecessor is alive
            try {
                this.references.getPredecessor().isLive();
                nodes.add(this.references.getPredecessor());
            } catch (RemoteException e) {
                this.references.removeReference(this.references.getPredecessor());
            }
            
        } else {
            nodes.add(potentialPredecessor); 
        }
        if (this.references.getSuccessor() != null) {
            nodes.add(this.references.getSuccessor());
        } else {
            nodes.add(potentialPredecessor);
        }
        //Add potential predecessor to successor and finger table.
        //Updates the predecessor of this node if it's a more appropriate predecessor.
        int predecessorID;
        if (this.references.getPredecessor() == null) {
            predecessorID = this.nodeID;
        } else {
            predecessorID = this.references.getPredecessor().nodeID;
        }
        boolean truePredecessor = this.references.addReferenceAndUpdatePredecessor(potentialPredecessor);  
        Set<Entry> entries = null;
        if (truePredecessor) {
            entries = this.entries.getEntriesInInterval(predecessorID, potentialPredecessor.getNodeID());
            if (this.nodeID != predecessorID) {
                removeCopies(this.entries.getEntriesInInterval(this.nodeID, predecessorID));
            }
            insertCopies(potentialPredecessor.getEntries());     
            this.references.getSuccessor().insertCopies(getEntries());
            Log.out("True predecessor found, retrieving entries for predecessor. Number of entries: " + entries.size());
        } else if (potentialPredecessor != null && potentialPredecessor.equals(this.references.getPredecessor())) {
            Set<Entry> transferEntries = this.entries.getEntriesInInterval(this.nodeID, potentialPredecessor.nodeID);
            transferEntries.removeAll(potentialPredecessor.getEntries());
            potentialPredecessor.insertCopies(transferEntries);
            this.entries.removeAll(transferEntries);
        }
        
        return new NodesAndEntries(nodes, entries);
    }
    
    /**
     * Get this node's host and port information.
     * 
     * @return Returns HostPort object for this node.
     */
    public final HostPort getNodeHostPort() {
        return this.nodeHostPort;
    }
    
    /**
     * Get this node's id.
     * 
     * @return Returns this nodes id.
     */
    public final int getNodeID() {
        return this.nodeID;
    }
    
    /**
     * Sets this node's host and port data.
     * 
     * @param nodeHostPort The host and port data for this node.
     */
    public final void setNodeHostPort(final HostPort nodeHostPort) {
        this.nodeHostPort = nodeHostPort;
    }
    
    /**
     * Sets this node's id.
     * 
     * @param nodeID The node id.
     */
    public final void setNodeID(final int nodeID) {
        this.nodeID = nodeID;
    }

    /**
     * 
     */
    public NodeReferences getNodeReferences() {
        return this.references;
    }

    @Override
    public void setSuccessor(Node successorNode) {
        this.references.setSuccessor(successorNode);
        System.out.println("successor of " + this.getNodeID() + " " + this.references.getSuccessor().getNodeID());
        
    }
    
    private void startTasks() {
        //Start Stabilize task.
        this.nodeTasks.scheduleWithFixedDelay(new Stabilize(this, this.references, this.entries), 
                                              STABILIZE_START_DELAY, 
                                              STABILIZE_INTERVAL, 
                                              TimeUnit.SECONDS);
        //Start Fix Finger Table task.
        this.nodeTasks.scheduleWithFixedDelay(new FixFingerTable(this, this.references), 
                                              FIX_FINGER_START_DELAY, 
                                              FIX_FINGER_INTERVAL, 
                                              TimeUnit.SECONDS);
    }
    
    /**
     * ThreadFactory used with executor service.
     */
    private static class NodeThreadFactory implements
            java.util.concurrent.ThreadFactory {

        private String executorServiceName;

        NodeThreadFactory(String executorServiceName) {
            this.executorServiceName = executorServiceName;
        }

        public Thread newThread(Runnable r) {
            Thread newThread = new Thread(r);
            newThread.setName(this.executorServiceName + "-" + newThread.getName());
            return newThread;
        }

    }
    
    /**
     * Prints message to log file.
     * 
     * @param the_message
     */
    private void messageToFile(String the_message) {
    	try {
    	Log.toFile(the_message, this.nodeID);
    	} catch (FileNotFoundException e) {
    		Log.err("could not create or find log file for node " + nodeID);
    	}
    }
    
    public static int getNodeIDFromKey(final int key, final int mValueOfChord) {
        return (int) (key % Math.pow(2, mValueOfChord));
    }

    @Override
    public void insertCopies(Set<Entry> entryCopiesForInsert) {
        this.entries.addAll(entryCopiesForInsert);        
    }

    @Override
    public void removeCopies(Set<Entry> entryCopiesForRemove) {
        this.entries.removeAll(entryCopiesForRemove);        
    }
    
    /**
     * @return Returns M value of chord.
     */
    public int getMValueOfChord() {
        return this.mValueOfChord;
    }

    @Override
    public void isLive() {
        //successful call of this method is proof node is Live.
    }

    @Override
    public Set<Entry> getEntries() {
        if (this.references.getPredecessor() != null) {
            return this.entries.getEntriesInInterval(this.references.getPredecessor().nodeID, this.nodeID);
        } else {
            return this.entries.getEntriesAsSet();
        }
    }
}
